import frappe
import datetime
import pika
import json
import time


#connection.close()

def get_connection():
    rb_settings = frappe.get_single('Rabbit MQ Settings')
    user = rb_settings.user
    password = rb_settings.get_password(fieldname="password", raise_exception=False)    
    
    credentials = pika.PlainCredentials(user, password)
    parameters = pika.ConnectionParameters(
        rb_settings.host, rb_settings.port, rb_settings.vhost, credentials
    )

    try:
        connection = pika.BlockingConnection(parameters)
        return connection
    except:
        pass

def get_channel(connection, exchange_name, queue_name, routing_key):
    channel = connection.channel()
    channel.exchange_declare(exchange=exchange_name, exchange_type='direct', durable=True) # Durable makes it survive broker restarts.
    channel.queue_declare(queue=queue_name, durable=True) # durable makes it survive broker restarts.
    channel.queue_bind(exchange=exchange_name, queue=queue_name, routing_key=routing_key)
        
    return channel

def publish_message(message_data, channel, exchange_name, routing_key):
    """Publishes a message to RabbitMQ."""
    try:
        channel.basic_publish(
            exchange=exchange_name,
            routing_key=routing_key,
            body=json.dumps(message_data),
            properties=pika.BasicProperties(
                delivery_mode=2,  # make message persistent
            ),
        )
        print(f"Published message: {message_data}")
    except pika.exceptions.AMQPConnectionError as e:
        print(f"Error connecting to RabbitMQ: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")

def consume_messages(queue_name):
    """Consumes messages from the RabbitMQ queue."""
    credentials = pika.PlainCredentials(rabbitmq_user, rabbitmq_password)
    parameters = pika.ConnectionParameters(
        rabbitmq_host, rabbitmq_port, rabbitmq_vhost, credentials
    )

    try:
        connection = pika.BlockingConnection(parameters)
        channel = connection.channel()

        channel.queue_declare(queue=queue_name, durable=True) # Ensure queue exists
        channel.basic_qos(prefetch_count=1) # process one message at a time

        def callback(ch, method, properties, body):
            try:
                message = json.loads(body)
                print(f"Received message: {message}")
                # Process the message here (e.g., update Frappe document, trigger other actions)
                ch.basic_ack(delivery_tag=method.delivery_tag) #acknowledge message processing.
            except json.JSONDecodeError:
                print("Invalid JSON received.")
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=False) #reject invalid message.
            except Exception as e:
                print(f"Error processing message: {e}")
                ch.basic_nack(delivery_tag=method.delivery_tag, requeue=True) #requeue message on error.

        channel.basic_consume(queue=queue_name, on_message_callback=callback)

        print("Waiting for messages...")
        channel.start_consuming()

    except pika.exceptions.AMQPConnectionError as e:
        print(f"Error connecting to RabbitMQ: {e}")
    except Exception as e:
        print(f"An unexpected error occurred: {e}")


def get_doctype_field_map():
    """
    
    """
    fm = frappe.qb.DocType("RB Field Map")
    fmd = frappe.qb.DocType("RB Field Map Detail")
    data = frappe.qb.from_(fm
    ).join(fmd
    ).on(fm.name == fmd.parent
    ).select(
        fm.reference_doctype,
        fm.queue_name,
        fm.rb_child_field_name,
        fmd.en_field_name,
        fmd.rb_field_name,
        fmd.default_value
    ).run(as_dict=1)

    doctype_fm_map = {}
    for d in data:
        fm_map = doctype_fm_map.setdefault(d.reference_doctype, {
            'queue_name':d.queue_name,
            'rb_child_field_name':d.rb_child_field_name,
            'fields': []
        })        
        fm_map['fields'].append(d)
    return doctype_fm_map

def get_message_data(doc):

    def convert_to_str(value):
        if isinstance(value, datetime.date) or isinstance(value, datetime.datetime):
            value = value.isoformat()
        return value

    message = {}
    doctype_fm_map = get_doctype_field_map()
    field_map = doctype_fm_map.get(doc.doctype)
    if field_map:
        fields = field_map.get('fields')        
        for field in fields:
            if field.en_field_name and field.rb_field_name:
                message[field.rb_field_name] = convert_to_str(doc.get(field.en_field_name))
            elif field.rb_field_name and field.default_value:
                message[field.rb_field_name] = field.default_value
            elif field.en_field_name and not field.rb_field_name:
                child_doctype = frappe.db.get_value("DocField", 
                    {
                        "parent": doc.doctype,   
                        "fieldname": field.en_field_name
                    },
                    "options"
                )
                if child_doctype:
                    child_field_map = doctype_fm_map.get(child_doctype)
                    child_fields = child_field_map.get('fields')
                    rb_child_field_name = child_field_map.get('rb_child_field_name')
                    message[rb_child_field_name] = []
                    rb_child_rows = message[rb_child_field_name]
                    rows = doc.get(field.en_field_name)
                    for row in rows:
                        rb_child_row = {}
                        for child_field in child_fields:
                            if child_field.en_field_name and child_field.rb_field_name:
                                rb_child_row[child_field.rb_field_name] = convert_to_str(row.get(child_field.en_field_name))
                            elif child_field.rb_field_name and child_field.default_value:
                                rb_child_row[child_field.rb_field_name] = child_field.default_value

                        item_code = row.get("item_code")
                        if item_code:
                            custom_内部镀种 = frappe.db.get_value("Item", item_code, "custom_内部镀种")
                            if custom_内部镀种:
                                rb_child_row["INSIDE_PLATING_NO"] = custom_内部镀种
                            custom_是否建模 = frappe.db.get_value("Item", item_code, "custom_是否建模")
                            if custom_是否建模 is not None:
                                rb_child_row["IS_MODE"] = custom_是否建模
                        if doc.doctype == "Sales Order":
                            # 返工标识
                            if rb_child_row.get("REWORK_FLAG") in ["是","返工"]:
                                rb_child_row["REWORK_FLAG"] = "1"
                            else:
                                rb_child_row["REWORK_FLAG"] = "0"
                            # 急品标识
                            if rb_child_row.get("IS_URGENT") in ["是", "急品"]:
                                rb_child_row["IS_URGENT"] = 1
                            else:
                                rb_child_row["IS_URGENT"] = 0
                            # 样品标识
                            if rb_child_row.get("IS_SAMPLE") in ["是", "样品"]:
                                rb_child_row["IS_SAMPLE"] = 1
                            else:
                                rb_child_row["IS_SAMPLE"] = 0
                            # 转换pcs数为重量
                            rb_child_row["WEIGHT"] = qty_to_kg_if_pcs(rb_child_row["EN_ITEM"], rb_child_row["WEIGHT"], rb_child_row["UOM"])

                        if doc.doctype == "Purchase Receipt":
                            minimum_packing_qty = frappe.db.get_value("Item", item_code, "min_pack_qty")
                            if minimum_packing_qty is not None:
                                rb_child_row["minimumPackingQty"] = minimum_packing_qty
                        
                        if doc.doctype == "Delivery Note":
                            # 转换pcs数为重量
                            rb_child_row["QTY"] = qty_to_kg_if_pcs(item_code, rb_child_row["QTY"], rb_child_row["UOM"])
                            # 下发时固定为kg, 生产只看kg
                            rb_child_row["UOM"] = "kg"
                        rb_child_rows.append(dict(rb_child_row))
            if doc.doctype == "Sales Order":
                customer_code = doc.get("customer")
                if customer_code:
                    group_flag = frappe.db.get_value("Customer", customer_code, "custom_理货标签打印类型")
                if group_flag:
                    message["GROUP_FLAG"] = group_flag
    return message

def sales_order_send_mq_message(doc, method=None):
    if not doc.custom_mq发送标识 or doc.custom_mq发送标识 == "未发送":
        exchange_name = "/sc"
        queue_name = "en_sd_order"
        routing_key = "en_sd_order"
        try:
            connection = get_connection()
            channel = get_channel(connection, exchange_name, queue_name, routing_key)
            message_data = get_message_data(doc)
            publish_message(message_data, channel, exchange_name, routing_key)

            doc.custom_mq发送标识 = "已发送"
            doc.save(ignore_permissions=True)

            frappe.get_doc({
                "doctype": "RB Sync Log",
                "reference_doctype": doc.doctype,
                "docname": doc.name,
                "sync_type": "发送",
                "sync_time": frappe.utils.now_datetime(),
                "sync_status": "成功",
                "custom_full_message": json.dumps(message_data, ensure_ascii=False)
            }).insert(ignore_permissions=True)

        except Exception as e:
            # 失败日志
            frappe.get_doc({
                "doctype": "RB Sync Log",
                "reference_doctype": doc.doctype,
                "docname": doc.name,
                "sync_type": "发送",
                "sync_time": frappe.utils.now_datetime(),
                "sync_status": "失败",
                "custom_full_message": str(e)
            }).insert(ignore_permissions=True)
            frappe.log_error(frappe.get_traceback(), "MQ 发送失败")
            raise

def sales_order_pull_mq_message():
    try:
        connection = get_connection()
        exchange_name="/sc"
        queue_name="en_sd_order_response"
        routing_key="en_sd_order_response"
        channel = get_channel(connection, exchange_name, queue_name, routing_key)

        while True:
            method_frame, header_frame, body = channel.basic_get(queue_name, auto_ack=False)

            if method_frame:
                try:
                    data = json.loads(body.decode())

                    sales_order_id = data.get("REQUEST_DATA").get("SALES_ORDER")
                    status = data.get("STATUS")

                    if sales_order_id and status:
                        so = frappe.get_doc("Sales Order", sales_order_id)
                        if status == "200":
                            so.custom_mq发送标识 = "成功"
                        else:
                            so.custom_mq发送标识 = "失败"
                        so.save(ignore_permissions=True)

                        frappe.get_doc({
                            "doctype": "RB Sync Log",
                            "reference_doctype": "Sales Order",
                            "docname": sales_order_id,
                            "sync_type": "接收",
                            "sync_time": frappe.utils.now_datetime(),
                            "sync_status": "成功",
                            "custom_full_message": json.dumps(data, ensure_ascii=False)
                        }).insert(ignore_permissions=True)

                    channel.basic_ack(delivery_tag=method_frame.delivery_tag)

                except Exception as e:
                    channel.basic_nack(delivery_tag=method_frame.delivery_tag, requeue=False)
                    frappe.get_doc({
                        "doctype": "RB Sync Log",
                        "reference_doctype": "Sales Order",
                        "docname": "Unknown",
                        "sync_type": "接收",
                        "sync_time": frappe.utils.now_datetime(),
                        "sync_status": "失败",
                        "custom_full_message": f"消息解析或处理异常：{frappe.get_traceback()}"
                    }).insert(ignore_permissions=True)
            else:
                # 队列为空，退出循环
                break

        connection.close()

    except Exception as e:
        frappe.get_doc({
            "doctype": "RB Sync Log",
            "reference_doctype": "Sales Order",
            "docname": "Unknown",
            "sync_type": "接收",
            "sync_time": frappe.utils.now_datetime(),
            "sync_status": "失败",
            "custom_full_message": f"连接或通道异常：{frappe.get_traceback()}"
        }).insert(ignore_permissions=True)


def sales_order_update_stock_qty():
    try:
        # 建立 MQ 连接
        connection = get_connection()
        exchange_name = "/sc"
        queue_name = "en_sd_order_update_stock_qty"
        routing_key = "en_sd_order_update_stock_qty"
        channel = get_channel(connection, exchange_name, queue_name, routing_key)

        while True:
            method_frame, header_frame, body = channel.basic_get(queue_name, auto_ack=False)

            if method_frame:
                try:
                    data = json.loads(body.decode())

                    order_no = data.get("ORDER_NO")
                    idx_list = list(data.get("CUSTOM_SEQ").values())
                    total_qty = data.get("QTY")

                    if order_no and idx_list and total_qty is not None:
                        so = frappe.get_doc("Sales Order", order_no)
                        updated_rows = []

                        # 构建 mapping：idx -> item
                        idx_item_map = {str(item.idx): item for item in so.items}
                        ordered_items = [idx_item_map.get(str(idx)) for idx in idx_list if str(idx) in idx_item_map]

                        remaining_qty = float(total_qty)

                        for i, item in enumerate(ordered_items):
                            if not item:
                                continue

                            # 转换成 float 类型
                            try:
                                current_val = float(item.custom_入库数量 or 0.0)
                            except (ValueError, TypeError):
                                current_val = 0.0

                            is_last = (i == len(ordered_items) - 1)

                            if not is_last:
                                max_can_add = max(0, float(item.qty) - current_val)
                                add_qty = min(remaining_qty, max_can_add)
                            else:
                                add_qty = remaining_qty

                            if add_qty > 0:
                                current_val += add_qty
                                item.custom_入库数量 = str(current_val)  

                                # 计算并更新可出库数量
                                try:
                                    out_qty = float(item.custom_已出库数量 or 0.0)
                                except (ValueError, TypeError):
                                    out_qty = 0.0
                                available_qty = current_val - out_qty
                                if available_qty < 0:
                                    available_qty = 0
                                item.custom_可出库数量 = str(available_qty)

                                remaining_qty -= add_qty
                                updated_rows.append(str(item.idx))

                        if updated_rows:
                            so.save(ignore_permissions=True)

                            frappe.get_doc({
                                "doctype": "RB Sync Log",
                                "reference_doctype": "Sales Order",
                                "docname": order_no,
                                "sync_type": "接收",
                                "sync_time": frappe.utils.now_datetime(),
                                "sync_status": "成功",
                                "custom_full_message": 
                                f"接收到的原始消息：{body.decode()}\n"
                                f"分配总数量：{total_qty}，已更新行：{', '.join(updated_rows)}，剩余未分配：{remaining_qty}"
                            }).insert(ignore_permissions=True)
                        else:
                            frappe.get_doc({
                                "doctype": "RB Sync Log",
                                "reference_doctype": "Sales Order",
                                "docname": order_no,
                                "sync_type": "接收",
                                "sync_time": frappe.utils.now_datetime(),
                                "sync_status": "无匹配行",
                                "custom_full_message": 
                                f"接收到的原始消息：{body.decode()}\n"
                                f"未找到匹配或可分配的行：{json.dumps(idx_list)}"
                            }).insert(ignore_permissions=True)

                    channel.basic_ack(delivery_tag=method_frame.delivery_tag)

                except Exception:
                    channel.basic_nack(delivery_tag=method_frame.delivery_tag, requeue=False)
                    frappe.get_doc({
                        "doctype": "RB Sync Log",
                        "reference_doctype": "Sales Order",
                        "docname": order_no if 'order_no' in locals() else "Unknown",
                        "sync_type": "接收",
                        "sync_time": frappe.utils.now_datetime(),
                        "sync_status": "失败",
                        "custom_full_message": f"处理异常：\n{frappe.get_traceback()}"
                    }).insert(ignore_permissions=True)
            else:
                break  # 队列为空，跳出循环

        connection.close()

    except Exception:
        frappe.get_doc({
            "doctype": "RB Sync Log",
            "reference_doctype": "Sales Order",
            "docname": "Unknown",
            "sync_type": "接收",
            "sync_time": frappe.utils.now_datetime(),
            "sync_status": "失败",
            "custom_full_message": f"连接异常：\n{frappe.get_traceback()}"
        }).insert(ignore_permissions=True)

def delivery_note_send_mq_message(doc, method=None):
    '''销售出库单发送mq消息到MES'''
    if not doc.custom_delivery_note_mq发送标识 or doc.custom_delivery_note_mq发送标识 == "未发送":
        exchange_name = "/sc"
        queue_name = "en_sd_delivery"
        routing_key = "en_sd_delivery"
        try:
            connection = get_connection()
            channel = get_channel(connection, exchange_name, queue_name, routing_key)
            message_data = get_message_data(doc)

            publish_message(message_data, channel, exchange_name, routing_key)

            doc.custom_delivery_note_mq发送标识 = "已发送"
            doc.save(ignore_permissions=True)

            frappe.get_doc({
                "doctype": "RB Sync Log",
                "reference_doctype": doc.doctype,
                "docname": doc.name,
                "sync_type": "发送",
                "sync_time": frappe.utils.now_datetime(),
                "sync_status": "成功",
                "custom_full_message": json.dumps(message_data, ensure_ascii=False)
            }).insert(ignore_permissions=True)

        except Exception as e:
            frappe.get_doc({
                "doctype": "RB Sync Log",
                "reference_doctype": doc.doctype,
                "docname": doc.name,
                "sync_type": "发送",
                "sync_time": frappe.utils.now_datetime(),
                "sync_status": "失败",
                "custom_full_message": str(e)
            }).insert(ignore_permissions=True)
            frappe.log_error(frappe.get_traceback(), "销售出库 MQ 发送失败")
            raise

def delivery_note_pull_mq_message():
    try:
        connection = get_connection()
        exchange_name = "/sc"
        queue_name = "en_sd_delivery_response"
        routing_key = "en_sd_delivery_response"
        channel = get_channel(connection, exchange_name, queue_name, routing_key)

        while True:
            method_frame, header_frame, body = channel.basic_get(queue_name, auto_ack=False)

            if method_frame:
                try:
                    data = json.loads(body.decode())

                    delivery_no = data.get("REQUEST_DATA", {}).get("DELIVERY_NO")
                    status = data.get("STATUS")

                    if delivery_no and status:
                        dn = frappe.get_doc("Delivery Note", delivery_no)
                        if status == "200":
                            dn.custom_delivery_note_mq发送标识 = "成功"
                        else:
                            dn.custom_delivery_note_mq发送标识 = "失败"
                        dn.save(ignore_permissions=True)

                        frappe.get_doc({
                            "doctype": "RB Sync Log",
                            "reference_doctype": "Delivery Note",
                            "docname": delivery_no,
                            "sync_type": "接收",
                            "sync_time": frappe.utils.now_datetime(),
                            "sync_status": "成功",
                            "custom_full_message": json.dumps(data, ensure_ascii=False)
                        }).insert(ignore_permissions=True)

                    channel.basic_ack(delivery_tag=method_frame.delivery_tag)

                except Exception as e:
                    channel.basic_nack(delivery_tag=method_frame.delivery_tag, requeue=False)
                    frappe.get_doc({
                        "doctype": "RB Sync Log",
                        "reference_doctype": "Delivery Note",
                        "docname": "Unknown",
                        "sync_type": "接收",
                        "sync_time": frappe.utils.now_datetime(),
                        "sync_status": "失败",
                        "custom_full_message": f"消息解析或处理异常：{frappe.get_traceback()}"
                    }).insert(ignore_permissions=True)
            else:
                break  # 队列为空，退出循环

        connection.close()

    except Exception as e:
        frappe.get_doc({
            "doctype": "RB Sync Log",
            "reference_doctype": "Delivery Note",
            "docname": "Unknown",
            "sync_type": "接收",
            "sync_time": frappe.utils.now_datetime(),
            "sync_status": "失败",
            "custom_full_message": f"连接或通道异常：{frappe.get_traceback()}"
        }).insert(ignore_permissions=True)

def pull_all_mq_messages():
    try:
        # 更新销售订单MQ发送标识
        sales_order_pull_mq_message()
        time.sleep(1)  # 等待 1 秒
        # 更新销售出库MQ发送标识
        delivery_note_pull_mq_message()
        time.sleep(1)
        # 更新销售订单入库数量
        sales_order_update_stock_qty()
        time.sleep(1)
        # 更新销售订单已出库数量和销售出库实际出库数量
        update_delivery_and_sales_order_qty()
        time.sleep(1)
        # 更新采购入库MQ发送标识
        purchase_receipt_pull_mq_message()
        time.sleep(1)
        # 更新销售订单收获批字段
        update_sales_order_contro_batch_mq_message()
    except Exception:
        frappe.get_doc({
            "doctype": "RB Sync Log",
            "reference_doctype": "System",
            "docname": "pull_all_mq_messages",
            "sync_type": "调度",
            "sync_time": frappe.utils.now_datetime(),
            "sync_status": "失败",
            "custom_full_message": f"统一任务异常：{frappe.get_traceback()}"
        }).insert(ignore_permissions=True)

def update_delivery_and_sales_order_qty():
    try:
        connection = get_connection()
        exchange_name = "/sc"
        queue_name = "en_dn_update_actual_qty"
        routing_key = "en_dn_update_actual_qty"
        channel = get_channel(connection, exchange_name, queue_name, routing_key)

        while True:
            method_frame, header_frame, body = channel.basic_get(queue_name, auto_ack=False)

            if not method_frame:
                break  # 没有新消息

            try:
                data = json.loads(body.decode())
                items = data.get("DELIVERY_NOTE_ITEMS", [])

                dn_doc_map = {}
                so_doc_map = {}

                for entry in items:
                    dn_no = entry.get("DELIVERY_NO")
                    idx = str(entry.get("IDX"))
                    real_qty = round(float(entry.get("REAL_QTY", 0)), 3)

                    if not dn_no or not idx:
                        continue

                    # 获取或缓存出库单
                    if dn_no not in dn_doc_map:
                        dn_doc_map[dn_no] = frappe.get_doc("Delivery Note", dn_no)
                    dn_doc = dn_doc_map[dn_no]

                    # 查找出库明细行
                    target_row = next((item for item in dn_doc.items if str(item.idx) == idx), None)
                    if not target_row:
                        continue

                    real_qty = kg_to_pcs_if_item_is_pcs(
                        target_row.item_code,
                        real_qty,
                        weight_per_unit_in_kg=False,   # 如果 weight_per_unit 用“千克”存，改为 True
                    )


                    # 累加出库明细字段
                    try:
                        current_qty = float(target_row.custom_实际出库数量 or 0)
                    except (ValueError, TypeError):
                        current_qty = 0
                    target_row.custom_实际出库数量 = str(round(current_qty + real_qty, 3))
                    '''
                    # 获取并缓存对应销售订单
                    so_no = target_row.against_sales_order
                    so_detail_id = target_row.so_detail

                    if so_no and so_detail_id:
                        if so_no not in so_doc_map:
                            so_doc_map[so_no] = frappe.get_doc("Sales Order", so_no)
                        so_doc = so_doc_map[so_no]

                        # 找到销售订单对应明细行
                        so_row = next((item for item in so_doc.items if item.name == so_detail_id), None)
                        if so_row:
                            try:
                                already_out = round(float(so_row.custom_已出库数量 or 0), 3)
                            except (ValueError, TypeError):
                                already_out = 0

                            new_out = already_out + real_qty
                            so_row.custom_已出库数量 = str(round(new_out, 3))

                            # 入库数量 - 已出库数量，不能小于 0
                            try:
                                in_qty = float(so_row.custom_入库数量 or 0)
                            except (ValueError, TypeError):
                                in_qty = 0

                            remaining = max(0, in_qty - new_out)
                            so_row.custom_可出库数量 = str(round(remaining, 3))
                '''
                # 保存所有出库单
                for doc in dn_doc_map.values():
                    doc.save(ignore_permissions=True)
                '''
                # 保存所有销售订单
                for doc in so_doc_map.values():
                    doc.save(ignore_permissions=True)
                '''
                # 写入成功日志
                frappe.get_doc({
                    "doctype": "RB Sync Log",
                    "reference_doctype": "Delivery Note",
                    "docname": ", ".join(dn_doc_map.keys()),
                    "sync_type": "接收",
                    "sync_time": frappe.utils.now_datetime(),
                    "sync_status": "成功",
                    "custom_full_message": f"接收到的原始消息：{body.decode()}"
                }).insert(ignore_permissions=True)

                channel.basic_ack(delivery_tag=method_frame.delivery_tag)

            except Exception:
                channel.basic_nack(delivery_tag=method_frame.delivery_tag, requeue=False)
                frappe.get_doc({
                    "doctype": "RB Sync Log",
                    "reference_doctype": "Delivery Note",
                    "docname": dn_no if 'dn_no' in locals() else "Unknown",
                    "sync_type": "接收",
                    "sync_time": frappe.utils.now_datetime(),
                    "sync_status": "失败",
                    "custom_full_message": f"处理异常：\n{frappe.get_traceback()}"
                }).insert(ignore_permissions=True)

        connection.close()

    except Exception:
        frappe.get_doc({
            "doctype": "RB Sync Log",
            "reference_doctype": "Delivery Note",
            "docname": "Unknown",
            "sync_type": "接收",
            "sync_time": frappe.utils.now_datetime(),
            "sync_status": "失败",
            "custom_full_message": f"连接异常：\n{frappe.get_traceback()}"
        }).insert(ignore_permissions=True)

def purchase_receipt_send_mq_message(doc, method=None):
    '''采购收货单发送mq消息到MES'''
    if not doc.custom_purchase_receipt_mq发送标识 or doc.custom_purchase_receipt_mq发送标识 == "未发送":
        exchange_name = "/sc"
        queue_name = "en_wm_purchase_receipt"
        routing_key = "en_wm_purchase_receipt"
        try:
            connection = get_connection()
            channel = get_channel(connection, exchange_name, queue_name, routing_key)
            message_data = get_message_data(doc)

            publish_message(message_data, channel, exchange_name, routing_key)

            doc.custom_purchase_receipt_mq发送标识 = "已发送"
            doc.save(ignore_permissions=True)

            frappe.get_doc({
                "doctype": "RB Sync Log",
                "reference_doctype": doc.doctype,
                "docname": doc.name,
                "sync_type": "发送",
                "sync_time": frappe.utils.now_datetime(),
                "sync_status": "成功",
                "custom_full_message": json.dumps(message_data, ensure_ascii=False)
            }).insert(ignore_permissions=True)

        except Exception as e:
            frappe.get_doc({
                "doctype": "RB Sync Log",
                "reference_doctype": doc.doctype,
                "docname": doc.name,
                "sync_type": "发送",
                "sync_time": frappe.utils.now_datetime(),
                "sync_status": "失败",
                "custom_full_message": str(e)
            }).insert(ignore_permissions=True)
            frappe.log_error(frappe.get_traceback(), "采购入库 MQ 发送失败")
            raise

def purchase_receipt_pull_mq_message():
    """更新采购入库单MQ标识"""
    try:
        connection = get_connection()
        exchange_name = "/sc"
        queue_name = "en_wm_purchase_receipt_response"
        routing_key = "en_wm_purchase_receipt_response"
        channel = get_channel(connection, exchange_name, queue_name, routing_key)

        while True:
            method_frame, header_frame, body = channel.basic_get(queue_name, auto_ack=False)

            if method_frame:
                try:
                    data = json.loads(body.decode())

                    receipt_no = data.get("REQUEST_DATA", {}).get("PURCHASE_RECEIPT_NO")
                    status = data.get("STATUS")

                    if receipt_no and status:
                        pr = frappe.get_doc("Purchase Receipt", receipt_no)
                        if status == "200":
                            pr.custom_purchase_receipt_mq发送标识 = "成功"
                        else:
                            pr.custom_purchase_receipt_mq发送标识 = "失败"
                        pr.save(ignore_permissions=True)

                        frappe.get_doc({
                            "doctype": "RB Sync Log",
                            "reference_doctype": "Purchase Receipt",
                            "docname": receipt_no,
                            "sync_type": "接收",
                            "sync_time": frappe.utils.now_datetime(),
                            "sync_status": "成功",
                            "custom_full_message": json.dumps(data, ensure_ascii=False)
                        }).insert(ignore_permissions=True)

                    channel.basic_ack(delivery_tag=method_frame.delivery_tag)

                except Exception as e:
                    channel.basic_nack(delivery_tag=method_frame.delivery_tag, requeue=False)
                    frappe.get_doc({
                        "doctype": "RB Sync Log",
                        "reference_doctype": "Purchase Receipt",
                        "docname": "Unknown",
                        "sync_type": "接收",
                        "sync_time": frappe.utils.now_datetime(),
                        "sync_status": "失败",
                        "custom_full_message": f"消息解析或处理异常：{frappe.get_traceback()}"
                    }).insert(ignore_permissions=True)
            else:
                break  # 队列为空，退出循环
    finally:
        connection.close()


def update_sales_order_contro_batch_mq_message():
    """更新销售订单明细中的 custom_contro_batch 字段"""
    try:
        connection = get_connection()
        exchange_name = "/sc"
        queue_name = "en_sd_order_response_control_batch"
        routing_key = "en_sd_order_response_control_batch"
        channel = get_channel(connection, exchange_name, queue_name, routing_key)

        while True:
            method_frame, header_frame, body = channel.basic_get(queue_name, auto_ack=False)

            if method_frame:
                try:
                    data = json.loads(body.decode())

                    sales_order_no = data.get("SALES_ORDER_NO")
                    batch_list = data.get("CONTROL_BATCH_LIST", [])

                    updated = 0
                    if sales_order_no and batch_list:
                        so = frappe.get_doc("Sales Order", sales_order_no)

                        for row in batch_list:
                            idx = int(row.get("IDX"))
                            control_batch = row.get("CONTROL_BATCH")

                            for item in so.items:
                                if item.idx == idx:
                                    item.custom_control_batch = control_batch
                                    updated += 1
                                    break

                        if updated > 0:
                            so.save(ignore_permissions=True)

                        frappe.get_doc({
                            "doctype": "RB Sync Log",
                            "reference_doctype": "Sales Order",
                            "docname": sales_order_no,
                            "sync_type": "接收",
                            "sync_time": frappe.utils.now_datetime(),
                            "sync_status": "成功",
                            "custom_full_message": json.dumps(data, ensure_ascii=False)
                        }).insert(ignore_permissions=True)

                    channel.basic_ack(delivery_tag=method_frame.delivery_tag)

                except Exception as e:
                    channel.basic_nack(delivery_tag=method_frame.delivery_tag, requeue=False)
                    frappe.get_doc({
                        "doctype": "RB Sync Log",
                        "reference_doctype": "Sales Order",
                        "docname": "Unknown",
                        "sync_type": "接收",
                        "sync_time": frappe.utils.now_datetime(),
                        "sync_status": "失败",
                        "custom_full_message": f"消息解析或处理异常：{frappe.get_traceback()}"
                    }).insert(ignore_permissions=True)
            else:
                break  # 队列为空，退出循环
    finally:
        connection.close()

def qty_to_kg_if_pcs(
    item_code: str,
    qty,
    uom: str,
    decimals: int = 3,
    *,
    weight_per_unit_in_kg: bool = False,
):
    from decimal import Decimal, ROUND_HALF_UP

    def _is_pos(v) -> bool:
        # 严格要求 > 0 才视为有效
        try:
            return _to_dec(v) > 0
        except Exception:
            return False

    def _to_dec(v: object) -> Decimal:
        if v in (None, ""):
            return Decimal("0")
        return Decimal(str(v).replace(",", ""))

    # 非 pcs：原始数量→float
    if not uom or str(uom).strip().lower() != "pcs":
        return float(_to_dec(qty))

    it = frappe.db.get_value(
        "Item",
        item_code,
        ["custom_actual_weight", "weight_per_unit"],
        as_dict=True,
    ) or {}

    unit_g = None
    # 1) 自定义单重（按克存）且 > 0 才用
    if _is_pos(it.get("custom_actual_weight")):
        unit_g = _to_dec(it["custom_actual_weight"])
    # 2) 标准单重（> 0）：默认按克；若实际以 kg 存，传 weight_per_unit_in_kg=True
    elif _is_pos(it.get("weight_per_unit")):
        w = _to_dec(it["weight_per_unit"])
        unit_g = w * Decimal("1000") if weight_per_unit_in_kg else w

    if unit_g is None:
        return float(_to_dec(qty))

    qty_dec = _to_dec(qty)
    weight_kg = (unit_g * qty_dec) / Decimal("1000")
    step = Decimal("1").scaleb(-decimals)  # 0.001
    # 这里直接转 float，避免后续 JSON 报错
    return float(weight_kg.quantize(step, rounding=ROUND_HALF_UP))

def kg_to_pcs_if_item_is_pcs(
    item_code: str,
    weight_kg,
    *,
    weight_per_unit_in_kg: bool = False,  # 若 weight_per_unit 实际以 kg 存则传 True
):
    """
    若物料的 stock_uom 为 PCS：
      pcs = floor( (weight_kg * 1000) / 单重[g] )  -> 返回 int
    否则：
      直接返回原始 weight_kg（float）

    单重优先取 Item.custom_actual_weight（克，且必须 > 0），否则回退到 Item.weight_per_unit（>0）。
    不再使用 weight_uom；若 weight_per_unit 实际以 kg 存，传 weight_per_unit_in_kg=True。
    """
    from decimal import Decimal, ROUND_FLOOR
    import frappe
    from functools import lru_cache

    def _to_dec(v) -> Decimal:
        if v in (None, ""):
            return Decimal("0")
        return Decimal(str(v).replace(",", "").strip())

    def _is_pos(v) -> bool:
        try:
            return _to_dec(v) > 0
        except Exception:
            return False

    @lru_cache(maxsize=4096)
    def _get_item_basics(code: str):
        it = frappe.db.get_value(
            "Item",
            code,
            ["stock_uom", "custom_actual_weight", "weight_per_unit"],
            as_dict=True,
        ) or {}
        return (
            (it.get("stock_uom") or ""),
            it.get("custom_actual_weight"),
            it.get("weight_per_unit"),
        )

    stock_uom, custom_actual_weight, weight_per_unit = _get_item_basics(item_code)

    # 只有物料单位是 PCS 才换算；其余返回原始 kg（float）
    if str(stock_uom).strip().lower() not in {"pcs"}:
        return float(_to_dec(weight_kg))

    # 计算单重（克）
    unit_g = None
    if _is_pos(custom_actual_weight):
        unit_g = _to_dec(custom_actual_weight)                           # 已是克
    elif _is_pos(weight_per_unit):
        w = _to_dec(weight_per_unit)
        unit_g = w * Decimal("1000") if weight_per_unit_in_kg else w     # kg→g 或直接当作 g

    # 没有有效单重：无法换算，返回原 kg（float）
    if unit_g is None or unit_g <= 0:
        return float(_to_dec(weight_kg))

    # kg -> pcs，向下取整
    kg_dec = _to_dec(weight_kg)
    pcs = (kg_dec * Decimal("1000")) / unit_g
    return int(pcs.to_integral_value(rounding=ROUND_FLOOR))